package com.cbh.binlog.core;

import com.cbh.binlog.common.SpringUtil;
import com.cbh.binlog.config.BinlogClientConfig;
import com.cbh.binlog.entity.BinaryLogEventCallback;
import com.cbh.binlog.entity.Column;
import com.cbh.binlog.entity.PointCallback;
import com.cbh.binlog.link.BinaryLogEventListener;
import com.cbh.binlog.link.PointStorage;
import com.cbh.binlog.link.factory.BinaryLogEventListenerFactory;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 核心启动入口
 *
 * @author ChenBiHui
 * @date 2025/1/16
 */
@Slf4j
@Data
public class Entrance {

    private static final List<ThreadPoolTaskExecutor> CHANNELS = new ArrayList<>();
    private static final AtomicInteger INDEX = new AtomicInteger(0);
    private static final ThreadPoolTaskExecutor DEFAULT = new ThreadPoolTaskExecutor();
    private final BinlogClientConfig config;

    static {
        DEFAULT.setCorePoolSize(0);
        DEFAULT.setMaxPoolSize(1);
        DEFAULT.setKeepAliveSeconds(20);
        DEFAULT.setThreadNamePrefix("default-work-");
        DEFAULT.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        DEFAULT.afterPropertiesSet();
    }

    public Entrance(BinlogClientConfig config) {
        this.config = config;
        schema();
        BinaryLogClient client = client();
        Thread thread = new Thread(() -> {
            try {
                client.connect();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }, "accept");
        thread.start();
    }

    public void schema() {
        String sql = "SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, ORDINAL_POSITION " +
                "FROM INFORMATION_SCHEMA.COLUMNS " +
                "ORDER BY TABLE_SCHEMA, TABLE_NAME, ORDINAL_POSITION";
        for (int i = 0; i < config.getLoad(); i++) {
            ThreadPoolTaskExecutor channel = new ThreadPoolTaskExecutor();
            channel.setCorePoolSize(0);
            channel.setMaxPoolSize(1);
            channel.setKeepAliveSeconds(60);
            channel.setThreadNamePrefix("amadeus-" + i + "-");
            channel.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            channel.afterPropertiesSet();
            CHANNELS.add(channel);
        }
        try (
                Connection connection = SchemaParser.dataSource().getConnection();
                PreparedStatement preparedStatement = connection.prepareStatement(sql);
                ResultSet resultSet = preparedStatement.executeQuery()
        ) {
            Set<String> modify = new HashSet<>();
            while (resultSet.next()) {
                String schema = resultSet.getString("TABLE_SCHEMA");
                String table = resultSet.getString("TABLE_NAME");
                modify.add(SchemaParser.compose(schema, table));
                Column column = new Column();
                column.setName(resultSet.getString("COLUMN_NAME"));
                column.setType(resultSet.getString("DATA_TYPE"));
                column.setPosition(resultSet.getInt("ORDINAL_POSITION") - 1);
                SchemaParser.save(schema, table, column);
            }
            for (String payload : modify) {
                SchemaParser.binding(payload, Entrance.match(config.getLoad()));
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    public BinaryLogClient client() {
        BinaryLogClient client = new BinaryLogClient(config.getHost(), config.getPort(), config.getUsername(), config.getPassword());
        EventDeserializer deserializer = new EventDeserializer();
        if (config.getOpenTrans()) {
            deserializer.setCompatibilityMode(
                    EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                    EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
            );
        }
        BinaryLogClient.EventListener listener;
        ThreadPoolTaskExecutor work = new ThreadPoolTaskExecutor();
        work.setCorePoolSize(0);
        work.setMaxPoolSize(1);
        work.setKeepAliveSeconds(120);
        work.setThreadNamePrefix("parse-");
        work.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        work.afterPropertiesSet();
        try {
            PointStorage storage = SpringUtil.getBean(PointStorage.class);
            PointCallback record = storage.record();
            if (record != null && StringUtils.hasText(record.getFilename()) && record.getNextPosition() != null) {
                client.setBinlogFilename(record.getFilename());
                client.setBinlogPosition(record.getNextPosition());
            }
            listener = new StorageEventListener(storage, record, work);
        } catch (Throwable ignored) {
            listener = new LatestEventListener(work);
        }
        client.setEventDeserializer(deserializer);
        client.registerEventListener(listener);
        client.setServerId(config.getServerId());
        return client;
    }

    static ThreadPoolTaskExecutor defaulted() {
        return DEFAULT;
    }

    static ThreadPoolTaskExecutor match(Integer load) {
        if (CHANNELS.isEmpty()) {
            return Entrance.defaulted();
        }
        return CHANNELS.get(Math.abs(INDEX.getAndIncrement() % load));
    }

    private static void deliver(EventType eventType, String schema, String table, List<List<Column>> parse) {
        BinaryLogEventListener start = BinaryLogEventListenerFactory.start();
        if (start == null) {
            return;
        }
        SchemaParser.channel(schema, table)
                .execute(() -> {
                    BinaryLogEventCallback callback = SchemaParser.callback(schema, table, parse);
                    start.process(eventType, schema, table, callback);
                });
    }

    public static class LatestEventListener implements BinaryLogClient.EventListener {

        private final ThreadPoolTaskExecutor work;

        public LatestEventListener(ThreadPoolTaskExecutor work) {
            this.work = work;
        }

        @Override
        public void onEvent(Event event) {
            if (BinaryLogEventListenerFactory.start() == null) {
                return;
            }
            work.execute(() -> {
                EventHeaderV4 header = event.getHeader();
                EventType eventType = header.getEventType();
                if (eventType == EventType.TABLE_MAP) {
                    TableMapEventData map = event.getData();
                    SchemaParser.load(map.getDatabase(), map.getTable(), map.getTableId());
                    return;
                }
                if (eventType == EventType.QUERY) {
                    QueryEventData query = event.getData();
                    SchemaParser.parse(query);
                    return;
                }
                if (EventType.isWrite(eventType)) {
                    WriteRowsEventData insert = event.getData();
                    Entrance.deliver(
                            eventType,
                            SchemaParser.findSchema(insert.getTableId()),
                            SchemaParser.findTable(insert.getTableId()),
                            SchemaParser.parse(insert)
                    );
                    return;
                }
                if (EventType.isUpdate(eventType)) {
                    UpdateRowsEventData update = event.getData();
                    Entrance.deliver(
                            eventType,
                            SchemaParser.findSchema(update.getTableId()),
                            SchemaParser.findTable(update.getTableId()),
                            SchemaParser.parse(update)
                    );
                    return;
                }
                if (EventType.isDelete(eventType)) {
                    DeleteRowsEventData delete = event.getData();
                    Entrance.deliver(
                            eventType,
                            SchemaParser.findSchema(delete.getTableId()),
                            SchemaParser.findTable(delete.getTableId()),
                            SchemaParser.parse(delete)
                    );
                }
            });
        }
    }

    public static class StorageEventListener implements BinaryLogClient.EventListener {

        private final PointStorage storage;
        private final PointCallback record;
        private final ThreadPoolTaskExecutor work;

        public StorageEventListener(PointStorage storage, PointCallback record, ThreadPoolTaskExecutor work) {
            this.storage = storage;
            this.record = record;
            this.work = work;
        }

        @Override
        public void onEvent(Event event) {
            if (BinaryLogEventListenerFactory.start() == null) {
                return;
            }
            work.execute(() -> {
                EventHeaderV4 header = event.getHeader();
                EventType eventType = header.getEventType();
                if (eventType == EventType.ROTATE) {
                    RotateEventData rotate = event.getData();
                    hold(rotate);
                }
                if (eventType == EventType.TABLE_MAP) {
                    TableMapEventData map = event.getData();
                    SchemaParser.load(map.getDatabase(), map.getTable(), map.getTableId());
                    hold(header);
                    return;
                }
                if (eventType == EventType.QUERY) {
                    QueryEventData query = event.getData();
                    SchemaParser.parse(query);
                    hold(header);
                    return;
                }
                if (EventType.isWrite(eventType)) {
                    WriteRowsEventData insert = event.getData();
                    Entrance.deliver(
                            eventType,
                            SchemaParser.findSchema(insert.getTableId()),
                            SchemaParser.findTable(insert.getTableId()),
                            SchemaParser.parse(insert)
                    );
                    hold(header);
                    return;
                }
                if (EventType.isUpdate(eventType)) {
                    UpdateRowsEventData update = event.getData();
                    Entrance.deliver(
                            eventType,
                            SchemaParser.findSchema(update.getTableId()),
                            SchemaParser.findTable(update.getTableId()),
                            SchemaParser.parse(update)
                    );
                    hold(header);
                    return;
                }
                if (EventType.isDelete(eventType)) {
                    DeleteRowsEventData delete = event.getData();
                    Entrance.deliver(
                            eventType,
                            SchemaParser.findSchema(delete.getTableId()),
                            SchemaParser.findTable(delete.getTableId()),
                            SchemaParser.parse(delete)
                    );
                    hold(header);
                }
            });
        }

        private void hold(RotateEventData rotate) {
            record.setFilename(rotate.getBinlogFilename());
            storage.hold(
                    new PointCallback()
                            .setFilename(rotate.getBinlogFilename())
                            .setNextPosition(rotate.getBinlogPosition())
            );
        }

        private void hold(EventHeaderV4 header) {
            storage.hold(
                    new PointCallback()
                            .setFilename(record.getFilename())
                            .setPosition(header.getPosition())
                            .setNextPosition(header.getNextPosition())
            );
        }
    }
}
